import org.apache.spark.sql.SparkSession
import org.apache.log4j.Logger

/**
  * Created by THINKPAD on 2017/7/26.
  * 输入需要转换的文件、经度位置、纬度位置、输出路径、转换方式
  * 02：火星转地球
  *       经度位置：999：逗号隔开的, 纬度位置是坐标位置
  * 84：地球转火星
  *
  *
  *
  */
object coorConvert {
  val ee = 0.00669342162296594323   // 扁率
  val a = 6378245.0  // 长半轴
  val pi = 3.1415926535897932384626  //π
  val log = Logger.getLogger(getClass.getName)

  def main(args: Array[String]) {
    if(args.length<5){
      System.err.println("You must pass the arguments:<sourceTable, lon, lat, dest, type>")
      System.exit(1)
    }

    val(table, lon,lat,dest,ty)=
      (args(0),args(1),args(2),args(3),args(4))
    val destOutName=dest.split('/').last

    val spark = SparkSession
      .builder()
      .master("local[8]")
      .appName("script")
      .config("spark.jars", "C:\\Codes\\IdeaProjects\\MachineLearning\\JiQiXueXi\\outJar\\JiQiXueXi.jar")
      .getOrCreate()
    log.info("-----------------------Start Spark Context-----------------------" +destOutName)

    val data =spark.sparkContext.textFile(table)
    var data1 = data
    if (ty == "02") //火星坐标转地球
    {
      data1 =   data.map(_.split("\t")).map(f=>{
        var lon1 = 0.1
        var lat1 = 0.1
        if (lon.toInt == 999){
          lon1 = f(lat.toInt).split(",")(0).toDouble
          lat1 = f(lat.toInt).split(",")(1).toDouble
        }else{
         lon1 = f(lon.toInt).toDouble
         lat1 = f(lat.toInt).toDouble }
        (lon1,lat1,f.mkString("\t"))
      })
//        过滤掉0,0
        .filter(f=>f._1 != 0 & f._2!=0).map(f=>{
        val str84 =  six( gcj02towgs84 (f._1 ,  f._2)._1) +"\t"+six(gcj02towgs84  (f._1 ,  f._2)._2)
        f._3+"\t"+  str84
      })
    }
    else if (ty == "84") //地球坐标转火星
    {
      data1 =   data.map(_.split("\t")).map(f=>{
        val lon1 = f(lon.toInt).toDouble
        val lat1 = f(lat.toInt).toDouble
        val str84 =  six( wgs84togcj02 ( lon1 , lat1)._1) +"\t"+six(wgs84togcj02  ( lon1 , lat1)._2)
        f.mkString("\t") +"\t" +  str84
      })
    }
    data1.coalesce(1).saveAsTextFile(dest)



  }
  def six(a:Double) = {
    BigDecimal(a).setScale(6, BigDecimal.RoundingMode.HALF_UP).toDouble
  }

  def wgs84togcj02(lng:Double, lat:Double)={
    //    WGS84转GCJ02(火星坐标系)
    //      :param lng:WGS84坐标系的经度
    //      :param lat:WGS84坐标系的纬度
    var dlat = transformlat(lng - 105.0, lat - 35.0)
    var dlng = transformlng(lng - 105.0, lat - 35.0)
    var radlat = lat / 180.0 * pi
    var    magic = math.sin(radlat)
    magic = 1 - ee * magic * magic
    val sqrtmagic = math.sqrt(magic)
    dlat = (dlat * 180.0) / ((a * (1 - ee)) / (magic * sqrtmagic) * pi)
    dlng = (dlng * 180.0) / (a / sqrtmagic * math.cos(radlat) * pi)
    val  mglat = lat + dlat
    val   mglng = lng + dlng
    (mglng,mglat)
  }

  def gcj02towgs84(lng :Double, lat:Double) = {
    //    GCJ02(火星坐标系)转GPS84
    //      :param lng:火星坐标系的经度
    //      :param lat:火星坐标系纬度
    // 0,0 不转换
    if (lng==0 && lat ==0) {
      (0.0,0.0)
    }else{

      var dlat = transformlat(lng - 105.0, lat - 35.0)
      var  dlng = transformlng(lng - 105.0, lat - 35.0)
      val   radlat = lat / 180.0 * math.Pi
      var    magic = math.sin(radlat)
      magic = 1 - ee * magic * magic
      val sqrtmagic = math.sqrt(magic)
      dlat = (dlat * 180.0) / ((a * (1 - ee)) / (magic * sqrtmagic) * pi)
      dlng = (dlng * 180.0) / (a / sqrtmagic * math.cos(radlat) * pi)
      val  mglat = lat + dlat
      val   mglng = lng + dlng
      (lng * 2 - mglng, lat * 2 - mglat)
    } }

  def transformlat(lng :Double, lat:Double) ={
    var ret = -100.0 + 2.0 * lng + 3.0 * lat + 0.2 * lat * lat +
      0.1 * lng * lat + 0.2 * math.sqrt(math.abs(lng))
    ret += (20.0 * math.sin(6.0 * lng * math.Pi) + 20.0 *
      math.sin(2.0 * lng * math.Pi)) * 2.0 / 3.0
    ret += (20.0 * math.sin(lat * math.Pi) + 40.0 *
      math.sin(lat / 3.0 * math.Pi)) * 2.0 / 3.0
    ret += (160.0 * math.sin(lat / 12.0 * math.Pi) + 320 *
      math.sin(lat * math.Pi / 30.0)) * 2.0 / 3.0
    ret
  }

  def transformlng(lng :Double, lat :Double) ={
    var ret = 300.0 + lng + 2.0 * lat + 0.1 * lng * lng +
      0.1 * lng * lat + 0.1 * math.sqrt(math.abs(lng))
    ret += (20.0 * math.sin(6.0 * lng *  math.Pi) + 20.0 *
      math.sin(2.0 * lng *  math.Pi)) * 2.0 / 3.0
    ret += (20.0 * math.sin(lng *  math.Pi) + 40.0 *
      math.sin(lng / 3.0 *  math.Pi)) * 2.0 / 3.0
    ret += (150.0 * math.sin(lng / 12.0 *  math.Pi) + 300.0 *
      math.sin(lng / 30.0 *  math.Pi)) * 2.0 / 3.0
    ret}

}
